// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2025 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Live execution client implementation for the OKX adapter.

use std::{
    future::Future,
    sync::{
        Mutex,
        atomic::{AtomicBool, Ordering},
    },
};

use anyhow::Context;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use futures_util::{StreamExt, pin_mut};
use nautilus_common::{
    live::{runner::get_exec_event_sender, runtime::get_runtime},
    messages::{
        ExecutionEvent, ExecutionReport as NautilusExecutionReport,
        execution::{
            BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
            GenerateOrderStatusReport, GeneratePositionReports, ModifyOrder, QueryAccount,
            QueryOrder, SubmitOrder, SubmitOrderList,
        },
    },
};
use nautilus_core::{MUTEX_POISONED, UUID4, UnixNanos, time::get_atomic_clock_realtime};
use nautilus_execution::client::{ExecutionClient, base::ExecutionClientCore};
use nautilus_live::execution::client::LiveExecutionClient;
use nautilus_model::{
    accounts::AccountAny,
    enums::{AccountType, OmsType, OrderType},
    events::{AccountState, OrderEventAny, OrderRejected, OrderSubmitted},
    identifiers::{AccountId, ClientId, InstrumentId, Venue},
    orders::Order,
    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
    types::{AccountBalance, MarginBalance},
};
use tokio::task::JoinHandle;

use crate::{
    common::{
        consts::{OKX_CONDITIONAL_ORDER_TYPES, OKX_VENUE},
        enums::{OKXInstrumentType, OKXMarginMode, OKXTradeMode},
    },
    config::OKXExecClientConfig,
    http::client::OKXHttpClient,
    websocket::{
        client::OKXWebSocketClient,
        messages::{ExecutionReport, NautilusWsMessage},
    },
};

#[derive(Debug)]
pub struct OKXExecutionClient {
    core: ExecutionClientCore,
    config: OKXExecClientConfig,
    http_client: OKXHttpClient,
    ws_private: OKXWebSocketClient,
    ws_business: OKXWebSocketClient,
    trade_mode: OKXTradeMode,
    exec_event_sender: Option<tokio::sync::mpsc::UnboundedSender<ExecutionEvent>>,
    started: bool,
    connected: AtomicBool,
    instruments_initialized: AtomicBool,
    ws_stream_handle: Option<JoinHandle<()>>,
    ws_business_stream_handle: Option<JoinHandle<()>>,
    pending_tasks: Mutex<Vec<JoinHandle<()>>>,
}

impl OKXExecutionClient {
    /// Creates a new [`OKXExecutionClient`].
    ///
    /// # Errors
    ///
    /// Returns an error if the client fails to initialize.
    pub fn new(core: ExecutionClientCore, config: OKXExecClientConfig) -> anyhow::Result<Self> {
        // Always use with_credentials which loads from env vars when config values are None
        let http_client = OKXHttpClient::with_credentials(
            config.api_key.clone(),
            config.api_secret.clone(),
            config.api_passphrase.clone(),
            config.base_url_http.clone(),
            config.http_timeout_secs,
            config.max_retries,
            config.retry_delay_initial_ms,
            config.retry_delay_max_ms,
            config.is_demo,
            config.http_proxy_url.clone(),
        )?;

        let account_id = core.account_id;
        let ws_private = OKXWebSocketClient::with_credentials(
            Some(config.ws_private_url()),
            config.api_key.clone(),
            config.api_secret.clone(),
            config.api_passphrase.clone(),
            Some(account_id),
            Some(20), // Heartbeat
        )
        .context("failed to construct OKX private websocket client")?;

        let ws_business = OKXWebSocketClient::with_credentials(
            Some(config.ws_business_url()),
            config.api_key.clone(),
            config.api_secret.clone(),
            config.api_passphrase.clone(),
            Some(account_id),
            Some(20), // Heartbeat
        )
        .context("failed to construct OKX business websocket client")?;

        let trade_mode = Self::derive_trade_mode(core.account_type, &config);

        Ok(Self {
            core,
            config,
            http_client,
            ws_private,
            ws_business,
            trade_mode,
            exec_event_sender: None,
            started: false,
            connected: AtomicBool::new(false),
            instruments_initialized: AtomicBool::new(false),
            ws_stream_handle: None,
            ws_business_stream_handle: None,
            pending_tasks: Mutex::new(Vec::new()),
        })
    }

    fn derive_trade_mode(account_type: AccountType, config: &OKXExecClientConfig) -> OKXTradeMode {
        let is_cross_margin = config.margin_mode == Some(OKXMarginMode::Cross);

        if account_type == AccountType::Cash {
            if !config.use_spot_margin {
                return OKXTradeMode::Cash;
            }
            return if is_cross_margin {
                OKXTradeMode::Cross
            } else {
                OKXTradeMode::Isolated
            };
        }

        if is_cross_margin {
            OKXTradeMode::Cross
        } else {
            OKXTradeMode::Isolated
        }
    }

    fn instrument_types(&self) -> Vec<OKXInstrumentType> {
        if self.config.instrument_types.is_empty() {
            vec![OKXInstrumentType::Spot]
        } else {
            self.config.instrument_types.clone()
        }
    }

    async fn refresh_account_state(&self) -> anyhow::Result<()> {
        let account_state = self
            .http_client
            .request_account_state(self.core.account_id)
            .await
            .context("failed to request OKX account state")?;

        self.core.generate_account_state(
            account_state.balances.clone(),
            account_state.margins.clone(),
            account_state.is_reported,
            account_state.ts_event,
        )
    }

    fn update_account_state(&self) -> anyhow::Result<()> {
        let runtime = get_runtime();
        runtime.block_on(self.refresh_account_state())
    }

    fn is_conditional_order(&self, order_type: OrderType) -> bool {
        OKX_CONDITIONAL_ORDER_TYPES.contains(&order_type)
    }

    fn submit_regular_order(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
        let order = cmd.order.clone();
        let ws_private = self.ws_private.clone();
        let trade_mode = self.trade_mode;

        self.spawn_task("submit_order", async move {
            ws_private
                .submit_order(
                    order.trader_id(),
                    order.strategy_id(),
                    order.instrument_id(),
                    trade_mode,
                    order.client_order_id(),
                    order.order_side(),
                    order.order_type(),
                    order.quantity(),
                    Some(order.time_in_force()),
                    order.price(),
                    order.trigger_price(),
                    Some(order.is_post_only()),
                    Some(order.is_reduce_only()),
                    Some(order.is_quote_quantity()),
                    None,
                )
                .await?;
            Ok(())
        });

        Ok(())
    }

    fn submit_conditional_order(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
        let order = cmd.order.clone();
        let trigger_price = order
            .trigger_price()
            .ok_or_else(|| anyhow::anyhow!("conditional order requires a trigger price"))?;
        let http_client = self.http_client.clone();
        let trade_mode = self.trade_mode;

        self.spawn_task("submit_algo_order", async move {
            http_client
                .place_algo_order_with_domain_types(
                    order.instrument_id(),
                    trade_mode,
                    order.client_order_id(),
                    order.order_side(),
                    order.order_type(),
                    order.quantity(),
                    trigger_price,
                    order.trigger_type(),
                    order.price(),
                    Some(order.is_reduce_only()),
                )
                .await?;
            Ok(())
        });

        Ok(())
    }

    fn cancel_ws_order(&self, cmd: &CancelOrder) -> anyhow::Result<()> {
        let ws_private = self.ws_private.clone();
        let command = cmd.clone();

        self.spawn_task("cancel_order", async move {
            ws_private
                .cancel_order(
                    command.trader_id,
                    command.strategy_id,
                    command.instrument_id,
                    Some(command.client_order_id),
                    Some(command.venue_order_id),
                )
                .await?;
            Ok(())
        });

        Ok(())
    }

    fn mass_cancel_instrument(&self, instrument_id: InstrumentId) -> anyhow::Result<()> {
        let ws_private = self.ws_private.clone();
        self.spawn_task("mass_cancel_orders", async move {
            ws_private.mass_cancel_orders(instrument_id).await?;
            Ok(())
        });
        Ok(())
    }

    fn spawn_task<F>(&self, description: &'static str, fut: F)
    where
        F: Future<Output = anyhow::Result<()>> + Send + 'static,
    {
        let runtime = get_runtime();
        let handle = runtime.spawn(async move {
            if let Err(e) = fut.await {
                tracing::warn!("{description} failed: {e:?}");
            }
        });

        let mut tasks = self.pending_tasks.lock().expect(MUTEX_POISONED);
        tasks.retain(|handle| !handle.is_finished());
        tasks.push(handle);
    }

    fn abort_pending_tasks(&self) {
        let mut tasks = self.pending_tasks.lock().expect(MUTEX_POISONED);
        for handle in tasks.drain(..) {
            handle.abort();
        }
    }
}

#[async_trait(?Send)]
impl ExecutionClient for OKXExecutionClient {
    fn is_connected(&self) -> bool {
        self.connected.load(Ordering::Acquire)
    }

    fn client_id(&self) -> ClientId {
        self.core.client_id
    }

    fn account_id(&self) -> AccountId {
        self.core.account_id
    }

    fn venue(&self) -> Venue {
        *OKX_VENUE
    }

    fn oms_type(&self) -> OmsType {
        self.core.oms_type
    }

    fn get_account(&self) -> Option<AccountAny> {
        self.core.get_account()
    }

    fn generate_account_state(
        &self,
        balances: Vec<AccountBalance>,
        margins: Vec<MarginBalance>,
        reported: bool,
        ts_event: UnixNanos,
    ) -> anyhow::Result<()> {
        self.core
            .generate_account_state(balances, margins, reported, ts_event)
    }

    fn start(&mut self) -> anyhow::Result<()> {
        if self.started {
            return Ok(());
        }

        self.started = true;

        // Spawn instrument bootstrap task
        let http_client = self.http_client.clone();
        let ws_private = self.ws_private.clone();
        let instrument_types = self.config.instrument_types.clone();

        get_runtime().spawn(async move {
            let mut all_instruments = Vec::new();
            for instrument_type in instrument_types {
                match http_client.request_instruments(instrument_type, None).await {
                    Ok(instruments) => {
                        if instruments.is_empty() {
                            tracing::warn!("No instruments returned for {instrument_type:?}");
                            continue;
                        }
                        http_client.cache_instruments(instruments.clone());
                        all_instruments.extend(instruments);
                    }
                    Err(e) => {
                        tracing::error!(
                            "Failed to request instruments for {instrument_type:?}: {e}"
                        );
                    }
                }
            }

            if all_instruments.is_empty() {
                tracing::warn!(
                    "Instrument bootstrap yielded no instruments; WebSocket submissions may fail"
                );
            } else {
                ws_private.cache_instruments(all_instruments);
                tracing::info!("OKX execution client instruments initialized");
            }
        });

        tracing::info!(
            client_id = %self.core.client_id,
            account_id = %self.core.account_id,
            account_type = ?self.core.account_type,
            trade_mode = ?self.trade_mode,
            instrument_types = ?self.config.instrument_types,
            use_fills_channel = self.config.use_fills_channel,
            is_demo = self.config.is_demo,
            http_proxy_url = ?self.config.http_proxy_url,
            ws_proxy_url = ?self.config.ws_proxy_url,
            "OKX execution client started"
        );
        Ok(())
    }

    fn stop(&mut self) -> anyhow::Result<()> {
        if !self.started {
            return Ok(());
        }

        self.started = false;
        self.connected.store(false, Ordering::Release);
        if let Some(handle) = self.ws_stream_handle.take() {
            handle.abort();
        }
        self.abort_pending_tasks();
        tracing::info!("OKX execution client {} stopped", self.core.client_id);
        Ok(())
    }

    fn submit_order(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
        let order = &cmd.order;

        if order.is_closed() {
            let client_order_id = order.client_order_id();
            tracing::warn!("Cannot submit closed order {client_order_id}");
            return Ok(());
        }

        let event = OrderSubmitted::new(
            self.core.trader_id,
            order.strategy_id(),
            order.instrument_id(),
            order.client_order_id(),
            self.core.account_id,
            UUID4::new(),
            cmd.ts_init,
            get_atomic_clock_realtime().get_time_ns(),
        );
        if let Some(sender) = &self.exec_event_sender
            && let Err(e) = sender.send(ExecutionEvent::Order(OrderEventAny::Submitted(event)))
        {
            tracing::warn!("Failed to send OrderSubmitted event: {e}");
        }

        let result = if self.is_conditional_order(order.order_type()) {
            self.submit_conditional_order(cmd)
        } else {
            self.submit_regular_order(cmd)
        };

        if let Err(e) = result {
            let rejected_event = OrderRejected::new(
                self.core.trader_id,
                order.strategy_id(),
                order.instrument_id(),
                order.client_order_id(),
                self.core.account_id,
                format!("submit-order-error: {e}").into(),
                UUID4::new(),
                cmd.ts_init,
                get_atomic_clock_realtime().get_time_ns(),
                false,
                false,
            );
            if let Some(sender) = &self.exec_event_sender
                && let Err(e) = sender.send(ExecutionEvent::Order(OrderEventAny::Rejected(
                    rejected_event,
                )))
            {
                tracing::warn!("Failed to send OrderRejected event: {e}");
            }
            return Err(e);
        }

        Ok(())
    }

    fn submit_order_list(&self, cmd: &SubmitOrderList) -> anyhow::Result<()> {
        tracing::warn!(
            "submit_order_list not yet implemented for OKX execution client (got {} orders)",
            cmd.order_list.orders.len()
        );
        Ok(())
    }

    fn modify_order(&self, cmd: &ModifyOrder) -> anyhow::Result<()> {
        let ws_private = self.ws_private.clone();
        let command = cmd.clone();

        self.spawn_task("modify_order", async move {
            ws_private
                .modify_order(
                    command.trader_id,
                    command.strategy_id,
                    command.instrument_id,
                    Some(command.client_order_id),
                    command.price,
                    command.quantity,
                    Some(command.venue_order_id),
                )
                .await?;
            Ok(())
        });

        Ok(())
    }

    fn cancel_order(&self, cmd: &CancelOrder) -> anyhow::Result<()> {
        self.cancel_ws_order(cmd)
    }

    fn cancel_all_orders(&self, cmd: &CancelAllOrders) -> anyhow::Result<()> {
        self.mass_cancel_instrument(cmd.instrument_id)
    }

    fn batch_cancel_orders(&self, cmd: &BatchCancelOrders) -> anyhow::Result<()> {
        let mut payload = Vec::with_capacity(cmd.cancels.len());

        for cancel in &cmd.cancels {
            payload.push((
                cancel.instrument_id,
                Some(cancel.client_order_id),
                Some(cancel.venue_order_id),
            ));
        }

        let ws_private = self.ws_private.clone();
        self.spawn_task("batch_cancel_orders", async move {
            ws_private.batch_cancel_orders(payload).await?;
            Ok(())
        });

        Ok(())
    }

    fn query_account(&self, _cmd: &QueryAccount) -> anyhow::Result<()> {
        self.update_account_state()
    }

    fn query_order(&self, cmd: &QueryOrder) -> anyhow::Result<()> {
        tracing::debug!(
            "query_order not implemented for OKX execution client (client_order_id={})",
            cmd.client_order_id
        );
        Ok(())
    }

    async fn connect(&mut self) -> anyhow::Result<()> {
        if self.connected.load(Ordering::Acquire) {
            return Ok(());
        }

        // Initialize exec event sender (must be done in async context after runner is set up)
        if self.exec_event_sender.is_none() {
            self.exec_event_sender = Some(get_exec_event_sender());
        }

        let instrument_types = self.instrument_types();

        if !self.instruments_initialized.load(Ordering::Acquire) {
            let mut all_instruments = Vec::new();
            for instrument_type in &instrument_types {
                let instruments = self
                    .http_client
                    .request_instruments(*instrument_type, None)
                    .await
                    .with_context(|| {
                        format!("failed to request OKX instruments for {instrument_type:?}")
                    })?;

                if instruments.is_empty() {
                    tracing::warn!("No instruments returned for {instrument_type:?}");
                    continue;
                }

                self.http_client.cache_instruments(instruments.clone());
                all_instruments.extend(instruments);
            }

            if !all_instruments.is_empty() {
                self.ws_private.cache_instruments(all_instruments);
            }
            self.instruments_initialized.store(true, Ordering::Release);
        }

        let Some(sender) = self.exec_event_sender.as_ref() else {
            tracing::error!("Execution event sender not initialized");
            anyhow::bail!("Execution event sender not initialized");
        };

        self.ws_private.connect().await?;
        self.ws_private.wait_until_active(10.0).await?;

        if self.ws_stream_handle.is_none() {
            let stream = self.ws_private.stream();
            let sender = sender.clone();
            let handle = tokio::spawn(async move {
                pin_mut!(stream);
                while let Some(message) = stream.next().await {
                    dispatch_ws_message(message, &sender);
                }
            });
            self.ws_stream_handle = Some(handle);
        }

        self.ws_business.connect().await?;
        self.ws_business.wait_until_active(10.0).await?;

        if self.ws_business_stream_handle.is_none() {
            let stream = self.ws_business.stream();
            let sender = sender.clone();
            let handle = tokio::spawn(async move {
                pin_mut!(stream);
                while let Some(message) = stream.next().await {
                    dispatch_ws_message(message, &sender);
                }
            });
            self.ws_business_stream_handle = Some(handle);
        }

        for inst_type in &instrument_types {
            tracing::info!(
                "Subscribing to channels for instrument type: {:?}",
                inst_type
            );
            self.ws_private.subscribe_orders(*inst_type).await?;

            if self.config.use_fills_channel
                && let Err(e) = self.ws_private.subscribe_fills(*inst_type).await
            {
                tracing::warn!("Failed to subscribe to fills channel ({inst_type:?}): {e}");
            }
        }

        self.ws_private.subscribe_account().await?;

        // Subscribe to algo orders on business WebSocket (OKX requires this endpoint)
        for inst_type in &instrument_types {
            if *inst_type != OKXInstrumentType::Option {
                self.ws_business.subscribe_orders_algo(*inst_type).await?;
            }
        }

        let account_state = self
            .http_client
            .request_account_state(self.core.account_id)
            .await
            .context("failed to request OKX account state")?;

        dispatch_account_state(account_state, sender);

        self.connected.store(true, Ordering::Release);
        tracing::info!(client_id = %self.core.client_id, "Connected");
        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        if !self.connected.load(Ordering::Acquire) {
            return Ok(());
        }

        self.abort_pending_tasks();
        self.http_client.cancel_all_requests();

        if let Err(e) = self.ws_private.close().await {
            tracing::warn!("Error while closing OKX private websocket: {e:?}");
        }

        if let Err(e) = self.ws_business.close().await {
            tracing::warn!("Error while closing OKX business websocket: {e:?}");
        }

        if let Some(handle) = self.ws_stream_handle.take() {
            handle.abort();
        }

        if let Some(handle) = self.ws_business_stream_handle.take() {
            handle.abort();
        }

        self.connected.store(false, Ordering::Release);
        tracing::info!(client_id = %self.core.client_id, "Disconnected");
        Ok(())
    }
}

#[async_trait(?Send)]
impl LiveExecutionClient for OKXExecutionClient {
    async fn generate_order_status_report(
        &self,
        cmd: &GenerateOrderStatusReport,
    ) -> anyhow::Result<Option<OrderStatusReport>> {
        let Some(instrument_id) = cmd.instrument_id else {
            tracing::warn!("generate_order_status_report requires instrument_id: {cmd:?}");
            return Ok(None);
        };

        let mut reports = self
            .http_client
            .request_order_status_reports(
                self.core.account_id,
                None,
                Some(instrument_id),
                None,
                None,
                false,
                None,
            )
            .await?;

        if let Some(client_order_id) = cmd.client_order_id {
            reports.retain(|report| report.client_order_id == Some(client_order_id));
        }

        if let Some(venue_order_id) = cmd.venue_order_id {
            reports.retain(|report| report.venue_order_id.as_str() == venue_order_id.as_str());
        }

        Ok(reports.into_iter().next())
    }

    async fn generate_order_status_reports(
        &self,
        cmd: &GenerateOrderStatusReport,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        let mut reports = Vec::new();

        if let Some(instrument_id) = cmd.instrument_id {
            let mut fetched = self
                .http_client
                .request_order_status_reports(
                    self.core.account_id,
                    None,
                    Some(instrument_id),
                    None,
                    None,
                    false,
                    None,
                )
                .await?;
            reports.append(&mut fetched);
        } else {
            for inst_type in self.instrument_types() {
                let mut fetched = self
                    .http_client
                    .request_order_status_reports(
                        self.core.account_id,
                        Some(inst_type),
                        None,
                        None,
                        None,
                        false,
                        None,
                    )
                    .await?;
                reports.append(&mut fetched);
            }
        }

        if let Some(client_order_id) = cmd.client_order_id {
            reports.retain(|report| report.client_order_id == Some(client_order_id));
        }

        if let Some(venue_order_id) = cmd.venue_order_id {
            reports.retain(|report| report.venue_order_id.as_str() == venue_order_id.as_str());
        }

        Ok(reports)
    }

    async fn generate_fill_reports(
        &self,
        cmd: GenerateFillReports,
    ) -> anyhow::Result<Vec<FillReport>> {
        let start_dt = nanos_to_datetime(cmd.start);
        let end_dt = nanos_to_datetime(cmd.end);
        let mut reports = Vec::new();

        if let Some(instrument_id) = cmd.instrument_id {
            let mut fetched = self
                .http_client
                .request_fill_reports(
                    self.core.account_id,
                    None,
                    Some(instrument_id),
                    start_dt,
                    end_dt,
                    None,
                )
                .await?;
            reports.append(&mut fetched);
        } else {
            for inst_type in self.instrument_types() {
                let mut fetched = self
                    .http_client
                    .request_fill_reports(
                        self.core.account_id,
                        Some(inst_type),
                        None,
                        start_dt,
                        end_dt,
                        None,
                    )
                    .await?;
                reports.append(&mut fetched);
            }
        }

        if let Some(venue_order_id) = cmd.venue_order_id {
            reports.retain(|report| report.venue_order_id.as_str() == venue_order_id.as_str());
        }

        Ok(reports)
    }

    async fn generate_position_status_reports(
        &self,
        cmd: &GeneratePositionReports,
    ) -> anyhow::Result<Vec<PositionStatusReport>> {
        let mut reports = Vec::new();

        // Query derivative positions (SWAP/FUTURES/OPTION) from /api/v5/account/positions
        if let Some(instrument_id) = cmd.instrument_id {
            let mut fetched = self
                .http_client
                .request_position_status_reports(self.core.account_id, None, Some(instrument_id))
                .await?;
            reports.append(&mut fetched);
        } else {
            for inst_type in self.instrument_types() {
                let mut fetched = self
                    .http_client
                    .request_position_status_reports(self.core.account_id, Some(inst_type), None)
                    .await?;
                reports.append(&mut fetched);
            }
        }

        // Query spot margin positions from /api/v5/account/balance
        // Spot margin positions appear as balance sheet items (liab/spotInUseAmt fields)
        let mut margin_reports = self
            .http_client
            .request_spot_margin_position_reports(self.core.account_id)
            .await?;

        if let Some(instrument_id) = cmd.instrument_id {
            margin_reports.retain(|report| report.instrument_id == instrument_id);
        }

        reports.append(&mut margin_reports);

        let _ = nanos_to_datetime(cmd.start);
        let _ = nanos_to_datetime(cmd.end);

        Ok(reports)
    }

    async fn generate_mass_status(
        &self,
        lookback_mins: Option<u64>,
    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
        tracing::warn!(
            "generate_mass_status not yet implemented (lookback_mins={lookback_mins:?})"
        );
        Ok(None)
    }
}

fn dispatch_ws_message(
    message: NautilusWsMessage,
    sender: &tokio::sync::mpsc::UnboundedSender<ExecutionEvent>,
) {
    match message {
        NautilusWsMessage::AccountUpdate(state) => dispatch_account_state(state, sender),
        NautilusWsMessage::PositionUpdate(report) => {
            dispatch_position_status_report(report, sender);
        }
        NautilusWsMessage::ExecutionReports(reports) => {
            tracing::debug!("Dispatching {} execution report(s)", reports.len());
            for report in reports {
                dispatch_execution_report(report, sender);
            }
        }
        NautilusWsMessage::OrderAccepted(event) => {
            dispatch_order_event(OrderEventAny::Accepted(event), sender);
        }
        NautilusWsMessage::OrderCanceled(event) => {
            dispatch_order_event(OrderEventAny::Canceled(event), sender);
        }
        NautilusWsMessage::OrderExpired(event) => {
            dispatch_order_event(OrderEventAny::Expired(event), sender);
        }
        NautilusWsMessage::OrderRejected(event) => {
            dispatch_order_event(OrderEventAny::Rejected(event), sender);
        }
        NautilusWsMessage::OrderCancelRejected(event) => {
            dispatch_order_event(OrderEventAny::CancelRejected(event), sender);
        }
        NautilusWsMessage::OrderModifyRejected(event) => {
            dispatch_order_event(OrderEventAny::ModifyRejected(event), sender);
        }
        NautilusWsMessage::OrderTriggered(event) => {
            dispatch_order_event(OrderEventAny::Triggered(event), sender);
        }
        NautilusWsMessage::OrderUpdated(event) => {
            dispatch_order_event(OrderEventAny::Updated(event), sender);
        }
        NautilusWsMessage::Error(e) => {
            tracing::warn!(
                "OKX websocket error: code={} message={} conn_id={:?}",
                e.code,
                e.message,
                e.conn_id
            );
        }
        NautilusWsMessage::Reconnected => {
            tracing::info!("OKX websocket reconnected");
        }
        NautilusWsMessage::Authenticated => {
            tracing::debug!("OKX websocket authenticated");
        }
        NautilusWsMessage::Deltas(_)
        | NautilusWsMessage::Raw(_)
        | NautilusWsMessage::Data(_)
        | NautilusWsMessage::FundingRates(_)
        | NautilusWsMessage::Instrument(_) => {
            tracing::debug!("Ignoring OKX websocket data message");
        }
    }
}

fn dispatch_account_state(
    state: AccountState,
    sender: &tokio::sync::mpsc::UnboundedSender<ExecutionEvent>,
) {
    if let Err(e) = sender.send(ExecutionEvent::Account(state)) {
        tracing::warn!("Failed to send account state: {e}");
    }
}

fn dispatch_position_status_report(
    report: PositionStatusReport,
    sender: &tokio::sync::mpsc::UnboundedSender<ExecutionEvent>,
) {
    let exec_report = NautilusExecutionReport::Position(Box::new(report));
    if let Err(e) = sender.send(ExecutionEvent::Report(exec_report)) {
        tracing::warn!("Failed to send position status report: {e}");
    }
}

fn dispatch_execution_report(
    report: ExecutionReport,
    sender: &tokio::sync::mpsc::UnboundedSender<ExecutionEvent>,
) {
    match report {
        ExecutionReport::Order(order_report) => {
            let exec_report = NautilusExecutionReport::OrderStatus(Box::new(order_report));
            if let Err(e) = sender.send(ExecutionEvent::Report(exec_report)) {
                tracing::warn!("Failed to send order status report: {e}");
            }
        }
        ExecutionReport::Fill(fill_report) => {
            let exec_report = NautilusExecutionReport::Fill(Box::new(fill_report));
            if let Err(e) = sender.send(ExecutionEvent::Report(exec_report)) {
                tracing::warn!("Failed to send fill report: {e}");
            }
        }
    }
}

fn dispatch_order_event(
    event: OrderEventAny,
    sender: &tokio::sync::mpsc::UnboundedSender<ExecutionEvent>,
) {
    if let Err(e) = sender.send(ExecutionEvent::Order(event)) {
        tracing::warn!("Failed to send order event: {e}");
    }
}

fn nanos_to_datetime(value: Option<UnixNanos>) -> Option<DateTime<Utc>> {
    value.map(|nanos| nanos.to_datetime_utc())
}

#[cfg(test)]
mod tests {
    use nautilus_common::messages::execution::{BatchCancelOrders, CancelOrder};
    use nautilus_core::UnixNanos;
    use nautilus_model::identifiers::{
        ClientId, ClientOrderId, InstrumentId, StrategyId, TraderId, VenueOrderId,
    };
    use rstest::rstest;

    #[rstest]
    fn test_batch_cancel_orders_builds_payload() {
        let trader_id = TraderId::from("TRADER-001");
        let strategy_id = StrategyId::from("STRATEGY-001");
        let client_id = ClientId::from("OKX");
        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
        let client_order_id1 = ClientOrderId::new("order1");
        let client_order_id2 = ClientOrderId::new("order2");
        let venue_order_id1 = VenueOrderId::new("venue1");
        let venue_order_id2 = VenueOrderId::new("venue2");

        let cmd = BatchCancelOrders {
            trader_id,
            client_id,
            strategy_id,
            instrument_id,
            cancels: vec![
                CancelOrder {
                    trader_id,
                    client_id,
                    strategy_id,
                    instrument_id,
                    client_order_id: client_order_id1,
                    venue_order_id: venue_order_id1,
                    command_id: Default::default(),
                    ts_init: UnixNanos::default(),
                },
                CancelOrder {
                    trader_id,
                    client_id,
                    strategy_id,
                    instrument_id,
                    client_order_id: client_order_id2,
                    venue_order_id: venue_order_id2,
                    command_id: Default::default(),
                    ts_init: UnixNanos::default(),
                },
            ],
            command_id: Default::default(),
            ts_init: UnixNanos::default(),
        };

        // Verify we can build the payload structure
        let mut payload = Vec::with_capacity(cmd.cancels.len());
        for cancel in &cmd.cancels {
            payload.push((
                cancel.instrument_id,
                Some(cancel.client_order_id),
                Some(cancel.venue_order_id),
            ));
        }

        assert_eq!(payload.len(), 2);
        assert_eq!(payload[0].0, instrument_id);
        assert_eq!(payload[0].1, Some(client_order_id1));
        assert_eq!(payload[0].2, Some(venue_order_id1));
        assert_eq!(payload[1].0, instrument_id);
        assert_eq!(payload[1].1, Some(client_order_id2));
        assert_eq!(payload[1].2, Some(venue_order_id2));
    }

    #[rstest]
    fn test_batch_cancel_orders_with_empty_cancels() {
        let cmd = BatchCancelOrders {
            trader_id: TraderId::from("TRADER-001"),
            client_id: ClientId::from("OKX"),
            strategy_id: StrategyId::from("STRATEGY-001"),
            instrument_id: InstrumentId::from("BTC-USDT.OKX"),
            cancels: vec![],
            command_id: Default::default(),
            ts_init: UnixNanos::default(),
        };

        let payload: Vec<(InstrumentId, Option<ClientOrderId>, Option<VenueOrderId>)> =
            Vec::with_capacity(cmd.cancels.len());
        assert_eq!(payload.len(), 0);
    }
}
